package com.atguigu.utils;

import com.atguigu.common.GmallConfig;
import com.atguigu.func.MyAnnonation;
import lombok.SneakyThrows;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;

// 将dws数据写进clickhouse里
public class MyClickhouseUtil {
    public static <T> SinkFunction<T> getSinkFunction(String sql) {  //让任意类型的对象都可以写进clickhouse
        return JdbcSink.sink(sql, new JdbcStatementBuilder<T>() {
                    @SneakyThrows
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t) {
                        Class<?> clz = t.getClass();
                        Field[] fields = clz.getDeclaredFields();//获得所有属性名，包括private修饰的

                        int offset=0;
                        for (int i = 0; i < fields.length; i++) {
                            Field field = fields[i];//获取下标为i的字段
                            field.setAccessible(true);

                            MyAnnonation myAnnonation = field.getAnnotation(MyAnnonation.class);
                            if (myAnnonation !=null){
                                offset--;
                                continue;

                            }
//                            String name = field.getName();//获得属性对应的属性名
                            Object fieldName = field.get(t);//获取属性对应的属性值
                            preparedStatement.setObject(i+1+offset,fieldName);

                        }
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(5)
                        .withBatchIntervalMs(1000L)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(GmallConfig.CLICKHOUSE_DRIVER)
                        .withUrl(GmallConfig.CLICKHOUSE_URL)
                        .build());


    }
}
